/******************************************************************************************************************
journal lite:
  1. journal records will be stored in files with postfix like .j, .j.0, .j.1, .j.2, ...
  2. journal records is defined as struct JournalRecord
  3. journal record contains only the range of a write operation, no data included (so called 'lite')
  4. journal file's format is defined as struct JournalFileHeader + n * struct JournalRecord
******************************************************************************************************************/

#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <dirent.h>
#include <inttypes.h> 
#include <stdbool.h>
#include <errno.h>
#include <string.h>
#include <ctype.h>
#include <fcntl.h>
#include <unistd.h>
#include <stddef.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>
#include "journal-lite.h"
#include "backend-fs.h"
#include "ursax.h"
#include "log.h"

#define JOURNAL_ROTATE_THRESHOLD (1024*1024*4)

#define JOURNAL_MAGIC 0x9d871bac

#define JournalLite2Chunk(j) container_of(j, struct Chunk, journal)

struct JournalFileHeader
{
	uint32_t magic;
	uint32_t padding; // struct JournalFileHeader size is 16 byte
	uint64_t version0;
};

const static char j_postfix[] = ".j";
int journal_force_flush = 0;

static int create_journal_file(uint32_t volumeid, uint32_t index, uint64_t version0)
{
	char path[PATH_MAX];
	int n = sprintf_chunk_path_filename(path, volumeid, index);
	strcpy( path+n, j_postfix );

	int fd = open(path, O_RDWR | O_CREAT | O_EXCL, 0644);
	if (unlikely(fd<=0))
	{
		LOG_ERROR("failed to create journal file '%s', %d: %s", path, errno, strerror(errno));
		return -1;
	}
	
	struct JournalFileHeader header;
	header.padding = 0;
	header.magic = JOURNAL_MAGIC;
	header.version0 = version0;
	int ret = write(fd, &header, sizeof(header));
	if (ret < 0)
	{
		LOG_ERROR("failed to initialize (write header) journal file '%s', %d: %s", path, errno, strerror(errno));
		close(fd);
		return -1;
	}
	return fd;
}

static int open_journal_file(uint32_t volumeid, uint32_t index, int generation)
{
	char path[PATH_MAX];
	int n = sprintf_chunk_path_filename(path, volumeid, index);
	strcpy( path+n, j_postfix );
	if (generation > 0)
	{
		const unsigned int len_jp = sizeof(j_postfix) - 1;
		snprintf(path + n + len_jp, sizeof(path) - n - len_jp, ".%u", generation);
	}

	int ret;
again:
	ret = open(path, O_RDWR, 0644);
	if (unlikely(ret <= 0))
	{
		LOG_INFO("failed to open journal file '%s', %d: %s", path, errno, strerror(errno));
		return -1;
	}
	
	int fd = ret;
	struct JournalFileHeader header;
	ret = pread(fd, &header, sizeof(header), 0);
	if (unlikely(ret < 0))
	{
		LOG_ERROR("failed to read header from journal file '%s', %d: %s", path, errno, strerror(errno));
		close(fd);
		return -1;
	}
	if (unlikely(header.magic != JOURNAL_MAGIC))
	{
		LOG_ERROR("journal magic doesn't match: got %08x, expected %08x, trying to unlink the file '%s'", header.magic, JOURNAL_MAGIC, path);
again2:
		close(fd);
		ret = unlink(path);
		if (unlikely(ret < 0))
		{
			LOG_ERROR("failed to unlink corrupted journal file '%s' due to %d: %s", path, errno, strerror(errno));
			return -1;
		}
		goto again;
	}

	uint32_t jfsize = file_size(fd);
	if (unlikely(jfsize > JOURNAL_ROTATE_THRESHOLD))
	{
		file_rotate(path);
	}
	else if (unlikely(jfsize == 0))
	{
		LOG_ERROR("trying to unlink the journal file");
		goto again2;
	}
	return fd;
}

static inline int open_create_journal_file(uint32_t volumeid, uint32_t index, uint64_t version0)
{
	int fd = open_journal_file(volumeid, index, 0);
	if (fd > 0)
		return fd;

	return create_journal_file(volumeid, index, version0);
}

static inline void journal_file_roate(uint32_t volumeid, uint32_t index)
{
	char path[PATH_MAX];
	sprintf_chunk_path_filename(path, volumeid, index);
	strcat(path, ".j");
	file_rotate(path);
}

void journal_lite_flush(struct JournalLite* j)
{
	if(0 == j->njournals){
		LOG_DEBUG("%s 0 == j->njournals return", __func__);
		return;
	}

	int fd;
	struct Chunk* c = JournalLite2Chunk(j);
again:
	fd = open_create_journal_file(c->volumeid, c->index, j->version0);
	if (fd <= 0)
		return;

	uint32_t jfsize = file_size(fd);
	struct JournalFileHeader header;
	pread(fd, &header, sizeof(header), 0);	// should have no errors
	uint32_t njournals = (jfsize - sizeof(header)) / sizeof(j->journals[0]);
	uint64_t version_n = header.version0 + njournals;
	if (version_n != j->version0)
	{
		LOG_INFO("there's a version gap between on-disk journal file (v%u) and " \
			"in-memory journal buffer (v%u), rotating", version_n, j->version0);
		close(fd);
		journal_file_roate(c->volumeid, c->index);
		goto again;
	}

	lseek(fd, 0, SEEK_END);
	uint32_t to_write = sizeof(j->journals[0]) * j->njournals;
	int ret = write(fd, j->journals, sizeof(j->journals[0]) * j->njournals);
	if (unlikely(ret < 0))
	{
		LOG_ERROR("failed to write buffered journals to file %d: %s", errno, strerror(errno));
	}
	else if (unlikely(ret < (int)to_write))
	{
		LOG_ERROR("partially written journal, the file may be corrupted");
		journal_file_roate(c->volumeid, c->index);
	}
	/** remove fsync to accelerate flush
	fsync(fd);
	*/
	close(fd);

	j->version0 += j->njournals;
	j->njournals = 0;
}

static inline void JRL_malloc(struct JRecordList* l, uint32_t size)
{
	l->njournals = size;
	l->r = (struct JournalRecord*)malloc(size * sizeof(l->r[0]));
}

static inline void JRL_realloc(struct JRecordList* l, uint32_t size)
{
	l->njournals = size;
	l->r = (struct JournalRecord*)realloc(l->r, size * sizeof(l->r[0]));
}

static int JR_cmp(const void* p1, const void* p2)
{
	const struct JournalRecord* a = (struct JournalRecord*)p1;
	const struct JournalRecord* b = (struct JournalRecord*)p2;
	return (int)(a->begin - b->begin);
}

#define BEGIN(i) (l->r[i].begin)
#define END(j) (l->r[j].begin + l->r[j].length)
#define LENGTH(a,b) (uint16_t)((b)-(a))
static void JRL_reduce(struct JRecordList* l)
{
	LOG_DEBUG("%s, njournals = %d", __func__, l->njournals);
	if (l->njournals <= 1) 
		return;

	qsort(l->r, l->njournals, sizeof(l->r[0]), &JR_cmp);

	uint32_t i, j;
	uint32_t begin0 = BEGIN(0), end0 = END(0);
	for (i=1, j=0; i < l->njournals; i++)
	{
		uint32_t begin = BEGIN(i), end = END(i);
		LOG_DEBUG("i=%d j=%d begin0=%d end0=%d begini=%d endi=%d", i, j, begin0, end0, begin, end);
		if (begin > end0)
		{
			j = i;
			begin0 = begin;
			end0 = end;
			continue;
		}
		if (end > end0)
		{
			l->r[j].length = LENGTH(begin0, end);
		}
		l->r[i].length = 0;   // mark [i] as 'to be deleted'
	}

	// deletes the items marked as 'to be deleted'
	for (i=j=0; i < l->njournals; ++i)
	{
		if (l->r[i].length > 0)
			l->r[j++] = l->r[i];
	}
	if (j < l->njournals)
		JRL_realloc(l, j);
}
#undef LENGTH
#undef END
#undef BEGIN

static inline void JRL_cat(struct JRecordList* l0, struct JRecordList* l1)
{
	uint32_t n0 = l0->njournals;
	uint32_t n1 = l1->njournals;
	JRL_realloc(l0, n0 + n1);
	memcpy(l0->r + n0, l1->r, n1 * sizeof(l1->r[0]));
}

static inline int JRL_from_memory(struct JRecordList* list, struct JournalLite* j, uint64_t since_version)
{
	uint64_t j_version_n = j->version0 + j->njournals;
	if (since_version >= j_version_n)
	{
		list->version0 = since_version;
		list->njournals = 0;
		list->r = NULL;
		LOG_INFO("%s(): since_version >= the latest journal record in memory, since_verison is %llu, \
				  j_version is %llu", __func__, since_version, j_version_n);
		return -1;
	}

	uint32_t n, offset;
	if (since_version >= j->version0)
	{
		list->version0 = since_version;
		n = (uint32_t)(j_version_n - since_version);
		offset = (uint32_t)(since_version - j->version0);
	}
	else // if (since_version < j->version0)
	{
		list->version0 = j->version0;
		n = j->njournals;
		offset = 0;
	}

	JRL_malloc(list, n);
	memcpy(list->r, j->journals + offset, n * sizeof(j->journals[0]));
	LOG_INFO("GET %u JRL FROM MEMORY, got since version %llu", n, list->version0);
	return 0;
}

static int journal_file_read(uint32_t volumeid, uint32_t index, int generation, uint64_t since_version, struct JRecordList* out_list)
{
	int fd = open_journal_file(volumeid, index, generation);
	if (fd < 0)
		return -1;
	
	struct JournalFileHeader header;
	int ret = read(fd, &header, sizeof(header));	// should have no errors
	if (ret < (int)sizeof(header))
	{
		LOG_ERROR("failed to read journal file header %d: %s", errno, strerror(errno));
		close(fd);
		return -1;
	}
	if (header.magic != JOURNAL_MAGIC)
	{
		LOG_ERROR("journal file magic error");
		close(fd);
		return -1;
	}

	struct JRecordList list;
	uint32_t jfsize = file_size(fd);
	uint32_t njournals = (jfsize - sizeof(header)) / sizeof(list.r[0]);
	int64_t skip = (int64_t)(since_version - header.version0);
	if (skip > 0)
	{
		njournals -= skip;
		list.version0 = since_version;
		lseek(fd, skip * sizeof(struct JournalRecord), SEEK_CUR);
	} else {
		list.version0 = header.version0;
	}

	JRL_malloc(&list, njournals);
	int to_read = njournals * sizeof(list.r[0]);
	ret = read(fd, list.r, to_read);
	if (ret < to_read)
	{
		LOG_ERROR("failed to read joural file due to %d: %s", errno, strerror(errno));
		JRL_free(&list);
		close(fd);
		return -1;
	}

	close(fd);
	*out_list = list;
	return 0;
}

#define JOURNAL_TOO_BIG (100*1024*1024 / sizeof(struct JournalRecord))
int journal_lite_read(struct JournalLite* j, uint64_t since_version, struct JRecordList* out_list)
{
	struct JRecordList list;
	struct JRecordList list1;
	struct Chunk* c = JournalLite2Chunk(j);
	int gen;
	int ret = JRL_from_memory(&list, j, since_version);
	if(ret < 0) {
		LOG_INFO("error since version >= last version");
		goto out_normal;
	}

	if (list.version0 == since_version)
	{
		LOG_DEBUG("all the requested journal records are in memory");
		goto out_normal;
	}

	for (gen=0; gen<99999; ++gen)
	{
		ret = journal_file_read(c->volumeid, c->index, gen, since_version, &list1);
		if (ret < 0)
		{
			JRL_free(&list);
			return -1;
		}
		if (list1.version0 + list1.njournals != list.version0)
		{
			LOG_DEBUG("list1.version0 %lld list1.njournals %ld list.version0 %lld", list1.version0, list1.njournals, list.version0);
			LOG_INFO("%s(chunk=%08x.%u, since_version=%llu): there's a version gap between on-disk journal " \
				"file of generation %d and %d, giving up", __func__, c->volumeid, c->index, since_version, gen, gen-1);
			JRL_free(&list1);
			JRL_free(&list);
			return -1;
		}

		list.version0 = list1.version0;
		JRL_cat(&list, &list1);
		JRL_free(&list1);
		if (list.version0 == since_version)
		{
out_normal:
			JRL_reduce(&list);
			*out_list = list;
			return 0;
		}

		if (list.njournals > JOURNAL_TOO_BIG)
		{
			JRL_reduce(&list);
		}
		if (list.njournals > JOURNAL_TOO_BIG)
		{
			LOG_WARN("%s(chunk=%08x.%u, since_version=%llu): journal bigger than bigger, giving up", 
				__func__, c->volumeid, c->index, since_version);
			JRL_free(&list);
			return -1;
		}
	}

	LOG_INFO("%s(chunk=%08x.%u, since_version=%llu): journal older than older, giving up",
		__func__, c->volumeid, c->index, since_version);
	JRL_free(&list);
	return -1;
}

static int do_remove_journal_file(struct dirent* entry, void* arg)
{
	char* prefix = (char*)arg;
	char* p = entry->d_name;
	while (*prefix != '\0')
		if (*p++ != *prefix++)
			return 0;

	char c = *p++;
	if (unlikely(c != '.' && c != '\0'))
	{
skip:
		LOG_WARN("file name '%s' seems to be a journal, but not strictly, skip deleting", entry->d_name);
		return 0;	// postfix is not a valid integer
	}

	if(c == '\0') // XXXX.j
		goto unlink;

	while (*p != '\0') // xxxx.j.1 xxxx.j.4 so on!!!
		if (!isdigit(*p++))
			goto skip;
unlink:
	LOG_DEBUG("unlinke journal file %s", entry->d_name);
	unlink(entry->d_name);
	return 0;
}

int journal_lite_remove(uint32_t volumeid, uint32_t index)
{
	char path[PATH_MAX], prefix[32];
	sprintf_chunk_path(path, volumeid, index);
	sprintf_chunk_filename(prefix, volumeid, index);
	strcat(prefix, ".j");

	dir_iterate(path, &do_remove_journal_file, prefix);

	return 0;
}


